package com.zc.study

import org.apache.commons.io.FileUtils
import org.apache.flink.api.common.accumulators.LongCounter
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.scala._

import collection.JavaConverters._
import org.apache.flink.configuration.Configuration
import org.apache.flink.core.fs.FileSystem.WriteMode

object DistributeCacheApp {

  def main(args: Array[String]): Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment
    val cacheFilePath = "Z:\\gitee\\study_imooc\\test-data\\flink\\distributeCache\\dict"

    env.registerCachedFile(cacheFilePath, "dictName")

    val dataSet = env.fromCollection(1 to 11)
//    val dataSet2 = env.fromElements(1,2,3)

    val handData = dataSet.map(new RichMapFunction[Int, Int] {
        override def open(parameters: Configuration): Unit = {
          val file = getRuntimeContext.getDistributedCache.getFile("dictName")
          val lines = FileUtils.readLines(file)
          for (line <- lines.asScala) {
            println(line)
          }
        }

        override def map(value: Int): Int = {
          value
        }
      }
    )

    val sinkFilePath = "Z:\\gitee\\study_imooc\\test-data\\flink\\sink-scala"
    handData.writeAsText(sinkFilePath,WriteMode.OVERWRITE)

    env.execute()

  }

}
